package cn.caplike.demo.kafka.producer.partition;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Objects;
import java.util.Properties;

public class ProducerDefaultPartitioner {

    public static void main(String[] args) {
        // 在控制台监听一个 topic: ./kafka-console-consumer.sh --bootstrap-server big-data-1:9092 big-data-2:9092 big-data-3:9092 --topic first

        // 配置
        final Properties properties = new Properties();

        // 连接集群 bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "big-data-1:9092,big-data-2:9092,big-data-3:9092");
        // 指定对应 key 和 value 的序列化类型 key.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 创建生产者对象
        try (final KafkaProducer<String, String/* "", hello */> producer = new KafkaProducer<>(properties)) {
            // 带回调的异步发送数据
            producer.send(new ProducerRecord<>("first",0, "", "caplike"), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (Objects.isNull(exception)) {
                        System.out.println("topic: " + metadata.topic());
                        System.out.println("partition: " + metadata.partition());
                    }
                }
            }); // 会自动创建 topic, 发送一条消息, 所有消费者都能收到
            // 如果分区不存在, 会阻塞?
        }

    }
}
